package com.hex.ds.hdtp.core.app.data.transfer.dto;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.util.StrUtil;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferFieldMapResponseDto;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfTemplatePo;
import lombok.Data;

import java.io.File;
import java.util.*;
import java.util.stream.Collectors;

@Data
public class DataTransferSqlStructureDto {

    private String dataTransferTaskPkId;

    private String dataTransferListPkId;

    private String sourceConnPkId;

    private String sourceDatabaseType;

    private String sourceDatabase;

    private String sourceTable;

    private String targetConnPkId;

    private String targetDatabaseType;

    private String targetDatabase;

    private String whereCondition;

    private String targetTable;

    private final String LINEFEED = "\r\n";

    private List<DateTime> bizDates;

    private List<String> partitionFieldNameList;

    private List<SourceConfTemplatePo> sourceConfList;

    private List<SourceConfTemplatePo> targetConfList;

    private List<TableTransferFieldMapResponseDto> metadatas;

    public String getSourceTableName() {
        return sourceDatabase + "." + sourceTable;
    }

    public String getTargetTableName() {
        return targetDatabase + "." + targetTable;
    }

    public String getDataTransferSqlFilePath() {
        return dataTransferTaskPkId + File.separator + getDataTransferSqlFileName();
    }

    public String getDataTransferSqlFileName() {
        return sourceDatabase + "." + sourceTable + "-" + targetDatabase + "." + targetTable + ".sql";
    }


    /**
     * @Method getTargetTableMetadata
     * @Param void
     * @Return java.lang.String
     * @Description 获取FlinkSQL目标表字段
     * @Author Yin.Yang
     * @Date 2024/3/8 16:10
     * @Version V1.0
     */
    public String getTargetTableMetadata() {
        StringBuilder targetTableMetadata = new StringBuilder();
        targetTableMetadata.append(LINEFEED);
        for (int i = 0; i < metadatas.size(); i++) {
            TableTransferFieldMapResponseDto metadata = metadatas.get(i);
            String fieldType = metadata.getTargetFieldType();
            if (StrUtil.isNotBlank(metadata.getTargetFieldLength()) && !"0".equals(metadata.getTargetFieldLength())) {
                targetTableMetadata.append(StrUtil.format("  `{}`  {}({})", metadata.getTargetFieldName(), fieldType, metadata.getTargetFieldLength()));
            } else {
                targetTableMetadata.append(StrUtil.format("  `{}`  {}", metadata.getTargetFieldName(), fieldType));
            }
            if (i < metadatas.size() - 1) {
                targetTableMetadata.append(",");
                targetTableMetadata.append(LINEFEED);
            }
        }
//        if (partitionFieldNameList != null && !partitionFieldNameList.isEmpty()) {
//            targetTableMetadata.append(",");
//            targetTableMetadata.append(LINEFEED);
//            for (int i = 0; i < partitionFieldNameList.size(); i++) {
//                String partitionFieldName = partitionFieldNameList.get(i);
//                targetTableMetadata.append(StrUtil.format("  `{}`  INTEGER", partitionFieldName));
//                if (i < partitionFieldNameList.size() - 1) {
//                    targetTableMetadata.append(",");
//                }
//                targetTableMetadata.append(LINEFEED);
//            }
//        }
        return targetTableMetadata.toString();
    }

    /**
     * @Method getSelectTableMetadata
     * @Param bizDate
     * @Return java.lang.String
     * @Description 获取FlinkSQL Select部分数据
     * @Author Yin.Yang
     * @Date 2024/3/11 18:12
     * @Version V1.0
     */
    public String getSelectTableMetadata(String bizDate) {
        StringBuilder selectTableMetadata = new StringBuilder();
//        String partionField = addPartition(bizDate);
        selectTableMetadata.append(LINEFEED);
        int size = metadatas.size();
        for (int i = 0; i < size; i++) {
            TableTransferFieldMapResponseDto metadata = metadatas.get(i);
            selectTableMetadata.append(formateSelectMetadata(metadata));
            // 仅在当前字段不是最后一个字段时添加逗号
            if (i < size - 1) {
                selectTableMetadata.append(",");
            }
            selectTableMetadata.append(LINEFEED);
        }
        // 当bizDate非空时，在最后一个字段后追加bizDate，注意不再额外添加逗号
//        if (StrUtil.isNotBlank(bizDate)) {
//            selectTableMetadata.append(",").append(LINEFEED).append("  ").append(bizDate);
//        }
        return selectTableMetadata.toString();
    }


    // 当前仅支持当个分区自动
    private String getPartitonField() {
        return partitionFieldNameList.get(0);
    }

    private String formateSelectMetadata(TableTransferFieldMapResponseDto tableTransferFieldMap) {
        String fieldType = tableTransferFieldMap.getTargetFieldType();
        if (StrUtil.isNotBlank(tableTransferFieldMap.getTargetFieldLength()) && !"0".equals(tableTransferFieldMap.getTargetFieldLength())) {
            return StrUtil.format("  cast(`{}` as {}({}))",
                    tableTransferFieldMap.getSourceFieldName(), fieldType,
                    tableTransferFieldMap.getTargetFieldLength());
        } else {
            return StrUtil.format("  cast(`{}` as {})",
                    tableTransferFieldMap.getSourceFieldName(), fieldType);
        }
    }

    // 当前仅支持当个分区自动
    private String addPartition(String bizDate) {
        List<TableTransferFieldMapResponseDto> collect = metadatas.stream().filter(a -> Objects.equals(a.getIsPartitionField(), "Y")).collect(Collectors.toList());
        if (CollectionUtil.isEmpty(collect)) {
            // 场景1：源表中没有字段可以作为分区字段，那么需要创建一个日期分区字段
            // 场景4：源表中存在与设置的分区字段名称相同（比如：data_td）,但该字段不能作为分区字段，目标表映射字段需要修改为其他名称（比如：data_td_src）
            return StrUtil.format("  {} as {}", getPartitonField(), bizDate);
        } else {
            TableTransferFieldMapResponseDto tableTransferFieldMap = collect.get(0);
            // 场景三: 源表中存在某个字段可以作为分区字段且字段名称和分区字段名称一样比如（data_td）
            if (tableTransferFieldMap.getSourceFieldName().equals(getPartitonField())) {
                return null;
            } else {
                // 场景2：源表中存在某个字段可以作为分区字段当字段字段名称和分区字段名称不一样
                return formateSelectMetadata(tableTransferFieldMap);
            }
        }
    }

    // 提取源库相关信息
    public Map getSourceConfMap() {
        Map<String, String> sourceConfMap = sourceConfList.stream().collect(Collectors.toMap(SourceConfTemplatePo::getConfName, SourceConfTemplatePo::getConfValue, (k, v) -> k));
        return sourceConfMap;
    }

    // 提取目标库相关信息
    public Map getTargetConfMap() {
        Map<String, String> targetConfMap = targetConfList.stream().collect(Collectors.toMap(SourceConfTemplatePo::getConfName, SourceConfTemplatePo::getConfValue, (k, v) -> k));
        return targetConfMap;
    }


    /**
     * @Method getCreateTableMetadata
     * @Param void
     * @Return java.lang.String
     * @Description 获取FlinkSQL Create部分数据
     * @Author Yin.Yang
     * @Date 2024/3/8 16:06
     * @Version V1.0
     */
    public String getCreateTableMetadata() {
        // 定义Flink不支持自定义长度的类型集合
        Set<String> unsupportedLengthTypes = new HashSet<>(Arrays.asList("INT", "INTEGER", "BIGINT", "SMALLINT", "TINYINT", "FLOAT", "DOUBLE", "VARCHAR", "CHAR", "BYTES"));
        return metadatas.stream().map(metadata -> {
                    boolean isUnsupportedLengthType = unsupportedLengthTypes.contains(metadata.getSourceFieldType().toUpperCase());
                    if (!isUnsupportedLengthType && StrUtil.isNotBlank(metadata.getSourceFieldLength()) && !"0".equals(metadata.getSourceFieldLength())) {
                        return StrUtil.format("  `{}`  {}({})", metadata.getSourceFieldName(), metadata.getSourceFieldType(), metadata.getSourceFieldLength());
                    } else {
                        return StrUtil.format("  `{}`  {}", metadata.getSourceFieldName(), metadata.getSourceFieldType());
                    }
                })
                .collect(Collectors.joining("," + LINEFEED, LINEFEED, LINEFEED));
    }

    /**
     * @Method getMysqlJDBCUrl
     * @Param oldMysqlJDBCUrl
     * @Param targetDatabase
     * @Return java.lang.String
     * @Description 特殊处理MysqlUrl，在端口号后面拼接库名
     * @Author Yin.Yang
     * @Date 2024/3/12 16:37
     * @Version V1.0
     */
    public String getMysqlJDBCUrl(String oldMysqlJDBCUrl, String targetDatabase) {
        if (StrUtil.isBlank(oldMysqlJDBCUrl) || StrUtil.isBlank(targetDatabase)) {
            return oldMysqlJDBCUrl;
        }
        int indexOfQuestionMark = oldMysqlJDBCUrl.indexOf("?");
        if (indexOfQuestionMark > -1) {
            String urlTemplate = "{}/{}{}";
            return StrUtil.format(urlTemplate, oldMysqlJDBCUrl.substring(0, indexOfQuestionMark),
                    targetDatabase, oldMysqlJDBCUrl.substring(indexOfQuestionMark));
        } else {
            String urlTemplate = "{}{}";
            return StrUtil.format(urlTemplate, oldMysqlJDBCUrl, "/" + targetDatabase);
        }
    }
}
